package wordcount;

import java.io.File;
import java.io.IOException;
import java.io.Reader;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;

import barrier.Barrier;

import com.ibm.tspaces.Field;
import com.ibm.tspaces.Tuple;
import com.ibm.tspaces.TupleSpace;
import com.ibm.tspaces.TupleSpaceException;

public class Master {
  public static final String   PARTITION_TAG = "partition";
  public static final String   DONE_TAG      = "done";
  public static final int      BUFFER_SIZE   = 10;

  private final File           file;
  private final TupleSpace     tspace;
  private final Tuple          clientFinishedTemplate;
  private final Tuple          wordCountTemplate;

  private int                  numberOfClients;
  private int                  numberOfClientsFinished;
  private Collection<String>   partitions;

  private Map<String, Integer> result;

  public Master(File file, int nClients) throws TupleSpaceException {
    this.file = file;
    tspace = new TupleSpace("wordCount", "localhost");
    tspace.deleteAll();
    numberOfClients = nClients;
    numberOfClientsFinished = 0;

    clientFinishedTemplate = new Tuple(DONE_TAG);
    wordCountTemplate = new Tuple(new Field(String.class), new Field(
        Integer.class));

    partitions = new java.util.LinkedList<String>();
    result = new java.util.TreeMap<String, Integer>();
  }

  public void partition(int nPartitions) throws IOException {
    long lengthInBytes = file.length();
    int partitionSize = (int) (lengthInBytes / nPartitions);
    
    System.err.println("partition: length of file: " + lengthInBytes);
    System.err.println("partition: # partitions: " + nPartitions);
    System.err.println("partition: partition size: " + partitionSize);

    Reader reader = new java.io.FileReader(file);
    char[] buf = new char[partitionSize];
    int read;

    while ((read = reader.read(buf)) != -1) {
      StringBuilder sb = new StringBuilder();
      sb.append(buf, 0, read);

      if (read == partitionSize && buf[read - 1] != ' ') {
        char c;
        while ((c = (char) reader.read()) != ' ')
          sb.append(c);
      }

      partitions.add(sb.toString());
    }
  }

  // particionar

  public void produceTasks() throws TupleSpaceException {
    for (String s : partitions)
      tspace.write(new Tuple(PARTITION_TAG, s));
  }

  // colocar no tspace

  public void agregate() throws TupleSpaceException {
    while (numberOfClientsFinished < numberOfClients) {
      Tuple wordCount;

      tspace.waitToTake(clientFinishedTemplate);
      numberOfClientsFinished++;

      while ((wordCount = tspace.take(wordCountTemplate)) != null) {
        String word = (String) wordCount.getField(0).getValue();
        int count = (Integer) wordCount.getField(1).getValue();

        Integer previousCount = result.get(word);

        if (previousCount == null)
          result.put(word, count);
        else
          result.put(word, count + previousCount);
      }
    }
  }

  // agregar enquanto não ler todos os dones.

  public Iterator<Entry<String, Integer>> getWordCount() {
    return result.entrySet().iterator();
  }

  public static void main(String[] args) throws IOException,
      TupleSpaceException {
    // args[0] => file name
    // args[1] => número de partições
    // args[2] => número de slaves
    
    TupleSpace bspace = new TupleSpace("barrier", "localhost");
    bspace.deleteAll();
    Barrier barrier = new Barrier("barrier", bspace);
    barrier.init();
    
    File file = new File(args[0]);
    int nClients = Integer.parseInt(args[2]);

    Master m = new Master(file, nClients);

    m.partition(Integer.parseInt(args[1]));
    m.produceTasks();
    System.err.println("Partições criadas!");
    barrier.barrier(nClients+1);
    long i = System.currentTimeMillis();
    m.agregate();
    long f = System.currentTimeMillis();
    
    System.err.println(f - i + "ms");

    Iterator<Entry<String, Integer>> it = m.getWordCount();
    while (it.hasNext()) {
      Entry<String, Integer> wordCount = it.next();
      System.out.printf("\t%d %s\n", wordCount.getValue(), wordCount.getKey());
    }
  }
}
